package data.platform.common.stream.aggregates;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import data.platform.common.domain.MetricValue;
import data.platform.common.stream.valueobjects.RuleMetricCondition;
import data.platform.common.stream.valueobjects.RuleState;
import data.platform.common.stream.valueobjects.WindowType;
import data.platform.common.stream.valueobjects.WorkingTimeWindow;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;

import java.text.DecimalFormat;
import java.text.DecimalFormatSymbols;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.*;
import java.util.stream.Collectors;

@Setter
@Getter
@NoArgsConstructor
@JsonIgnoreProperties
public class Rule {

    private String id;

    private String name;

    private RuleState ruleState;

    private Map<String, Set<String>> tags = new HashMap<>();

    private List<RuleMetricCondition> ruleMetricConditions = new ArrayList<>();

    private WindowType windowType;

    private int windowSize;


    // seconds
    private int waterMark = 0;
    //    时间段
    private List<WorkingTimeWindow> workingTimeWindows = new ArrayList<>();

    private static final ThreadLocal<DecimalFormat> TWO_DECIMAL_FORMAT = ThreadLocal.withInitial(() -> {
        // the following will ensure a dot ('.') as decimal separator
        DecimalFormatSymbols otherSymbols = new DecimalFormatSymbols(Locale.CHINA);
        return new DecimalFormat("#.##", otherSymbols);
    });

    public List<String> getMetrics() {
        return ruleMetricConditions.stream().map(RuleMetricCondition::getMetric).collect(Collectors.toList());
    }

    public LocalDateTime converTime(LocalTime localTime) {
        return LocalDate.now().atTime(localTime);
    }

    //    判断是否在此时间段集合内
    public boolean validateTimeSlot(LocalTime localTime) {
//       如果没有设置时间段,默认全部处理
        if(workingTimeWindows.size()==0){
            return true;
        }
        long count = workingTimeWindows.stream()
//               时间时间在时间段内
                .filter(t -> converTime(localTime).isBefore(converTime(t.getEndTime()).plusSeconds(waterMark)) && converTime(localTime).isAfter(converTime(t.getBeginTime())))
                .count();
        if (count > 0) {
            return true;
        } else {
            return false;
        }
    }

    public boolean validateMetric(String metric) {
        return ruleMetricConditions.stream()
                .filter(ruleMetricCondition -> ruleMetricCondition.getMetric().equals(metric))
                .findAny()
                .isPresent();
    }

    public boolean validateTags(Map<String, String> dataPointTags) {
        for (Map.Entry<String, Set<String>> entry : tags.entrySet()) {
            if (!dataPointTags.containsKey(entry.getKey())) {
                return false;
            }
            if (!entry.getValue().contains(dataPointTags.get(entry.getKey()))) {
                return false;
            }
        }
        return true;
    }

    public Map<Integer, WindowResult> apply(Map<String, Map<Integer, WindowAggregateItem>> metricCache) {
        // tagId -> result
        Map<Integer, WindowResult> comparisonResult = new HashMap<>();
        for (RuleMetricCondition ruleMetricCondition : ruleMetricConditions) {
            if (metricCache.containsKey(ruleMetricCondition.getMetric())) {
                for (Map.Entry<Integer, WindowAggregateItem> tagEntry : metricCache.get(ruleMetricCondition.getMetric()).entrySet()) {
                    Double value;
                    switch (ruleMetricCondition.getAggregatorFunctionType()) {
                        case MAX:
                            value = tagEntry.getValue().getMax();
                            break;
                        case MIN:
                            value = tagEntry.getValue().getMin();
                            break;
                        case SUM:
                            value = tagEntry.getValue().getSum();
                            break;
                        case COUNT:
                            value = Double.valueOf(tagEntry.getValue().getCount());
                            break;
                        default:
                            value = tagEntry.getValue().getAvg();
                            break;
                    }

                    value = Double.valueOf(TWO_DECIMAL_FORMAT.get().format(value));

                    boolean isHit = ruleMetricCondition.apply(value);
                    WindowResult windowResult = comparisonResult.computeIfAbsent(tagEntry.getKey(), k -> new WindowResult());
                    if (isHit) {
                        windowResult.getHit().compareAndSet(false, true);
                    }
                    windowResult.getMetricValue().put(ruleMetricCondition.getMetric(), value);
                }
            }
        }
        return comparisonResult;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        Rule rule = (Rule) o;
        return Objects.equals(id, rule.id);
    }

    @Override
    public int hashCode() {
        return Objects.hash(id);
    }

    @Override
    public String toString() {
        return "Rule{" +
                "id='" + id + '\'' +
                ", name='" + name + '\'' +
                ", ruleState=" + ruleState +
                ", tags=" + tags +
                ", ruleMetricConditions=" + ruleMetricConditions +
                ", windowType=" + windowType +
                ", windowSize=" + windowSize +
                ", waterMark=" + waterMark +
                ", workingTimeWindows=" + workingTimeWindows +
                '}';
    }
}
